package com.unis.utils

import com.alibaba.fastjson.JSON
import kafka.utils.ZkUtils

object KafkaZkUtils {
  /**
    * 将kafka topic的主体名称、分区信息，以及第一次读取的初始化值存入
    * 保存偏移量的文件内,这个文件第一次读取时不存在，就手动创建
    * @param zkUtil
    * @param path
    * @param topic
    */
  def setData(zkUtil:ZkUtils,path:String,topic: String): Unit= {
    val jsonData =zkUtil.readDataMaybeNull("/brokers/topics/"+topic)._1.get
    val json =JSON.parseObject(JSON.parseObject(jsonData).get("partitions").toString).keySet()
    import scala.collection.JavaConversions._
    for (key <- json) {
      zkUtil.updatePersistentPath(path,s"$topic,$key,0")
    }
  }


}
